package com.king.first.app.ad;

import com.king.first.bean.AdsClickLog;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;


public class Project_Ads_Click {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env
                .readTextFile("input/AdClickLog.csv")
                .map(line -> {
                    String[] datas = line.split(",");
                    return new AdsClickLog(
                            Long.valueOf(datas[0]),
                            Long.valueOf(datas[1]),
                            datas[2],
                            datas[3],
                            Long.valueOf(datas[4]));
                })
                .map(log -> Tuple2.of(Tuple2.of(log.getProvince(), log.getAdId()), 1L))
                .returns(Types.TUPLE(Types.TUPLE(Types.STRING, Types.LONG), Types.LONG))
                .keyBy(value -> value.f0)
                .sum(1)
                .print("省份-广告");

        env.execute();
    }

    public static void main1(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        SingleOutputStreamOperator<AdsClickLog> adsClickLogStream = env
                .readTextFile("input/AdClickLog.csv")
                .map(line -> {
                    String[] datas = line.split(",");
                    return new AdsClickLog(
                            Long.valueOf(datas[0]),
                            Long.valueOf(datas[1]),
                            datas[2],
                            datas[3],
                            Long.valueOf(datas[4]));
                });
        SingleOutputStreamOperator<Tuple2<Long, Integer>> streamOperator = adsClickLogStream.map(new MapFunction<AdsClickLog, Tuple2<Long, Integer>>() {
            @Override
            public Tuple2<Long, Integer> map(AdsClickLog adsClickLog) throws Exception {
                return Tuple2.of(adsClickLog.getAdId(), 1);
            }
        });

        streamOperator.keyBy(key -> key.f0).sum(1).print();

        env.execute();

    }
}
